package com.bigdata.schaduler.domain;

import com.bigdata.schaduler.util.Param;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.common.enums.ProcessExecutionTypeEnum;
import org.apache.dolphinscheduler.dao.entity.DagData;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;

import java.util.ArrayList;
import java.util.List;

@Data
public class ProcessDefineParam {

  @Param
  private String name;
  /**
   * 工作流编码，创建时生成编码并返回
   */
  @Param(required = false, note = "编辑操作true、创建操作false")
  private Long code;
  private String description;
  /**
   * 工作流全局参数
   */
  private List<Property> globalParams;
  /**
   * 租户，默认admin，操作系统操作用户
   */
  @Param
  private String tenantCode = "admin";
  /**
   * 超时时间，分钟。提交processInstance成功后，进行实例超时检测，超时则告警。
   * 超时从流程实例超时队列中移除 -> 并产生一个实例告警事件 -> 插入一条流程超时ALTER数据
   */
  private int timeout = 0;
  /**
   * 执行方式，同processDefinitionCode和processDefinitionVersion下的实例是否可以并行执行
   * master[slot], 通过command id % masterCount == slot扫描10个Command进行处理
   * Command TO ProcessInstance，判断ProcessInstance执行方式
   * 并行，生成ProcessInstance成功，并移除Command
   * 串行等待，更新ProcessInstance为串行等待状态，没有运行中的状态则改为执行中并返回；否则返回NULL，并且不移除Command，等待下一次扫描并执行Command
   * 串行丢弃，更新ProcessInstance为stop状态，不做任何返回，下次扫描不到该命令
   * 串行优先级，最好别用，会block运行中的流程，然后让当前流程先运行
   */
  private ProcessExecutionTypeEnum executionType = ProcessExecutionTypeEnum.PARALLEL;

  /**
   * 任务节点定义
   */
  @Param
  private List<TaskDefineParam> taskDefinitionJson;
  /**
   * 任务前后关系
   */
  @Param
  private List<TaskRelationParam> taskRelationJson;
  private List<TaskLocationParam> locations;
  private JSONObject otherParamsJson;

  /**
   * 查询流程定义，并对其进行编辑；或者接收前端传过来已编辑的DagData，转换后调用Dolphin编辑接口
   *
   * @param dagData 通过流程定义CODE，查询出来的流程定义完整数据；前端传过来的完整DagData
   */
  public void convertDagDataToProcessDefineParam(DagData dagData) {
    Assert.notNull(dagData, "convertDagDataToProcessDefineParam, processDefinition is NULL");
    ProcessDefinition processDefinition = dagData.getProcessDefinition();
    setProcessDefineParam(processDefinition);

    String locationJSONString = processDefinition.getLocations();
    setLocationJSONString(locationJSONString);

    TaskDefineParam taskDefineParam = new TaskDefineParam();
    this.setTaskDefinitionJson(taskDefineParam.setTaskDefineParamList(dagData.getTaskDefinitionList()));

    TaskRelationParam taskRelationParam = new TaskRelationParam();
    this.setTaskRelationJson(taskRelationParam.setTaskRelationParamList(dagData.getProcessTaskRelationList()));
  }

  public void setProcessDefineParam(ProcessDefinition processDefinition) {
    String tenantCode = processDefinition.getTenantCode();
    if (StringUtils.isEmpty(tenantCode)) {
      tenantCode = "admin";
    }

    this.setCode(processDefinition.getCode());
    this.setDescription(processDefinition.getDescription());
    this.setName(processDefinition.getName());
    this.setGlobalParams(processDefinition.getGlobalParamList());
    this.setTimeout(processDefinition.getTimeout());
    this.setExecutionType(processDefinition.getExecutionType());
    this.setTenantCode(tenantCode);
  }

  public void setLocationJSONString(String locationJSONString) {
    if (StrUtil.isEmpty(locationJSONString)) {
      return;
    }

    JSONArray jsonArray = new JSONArray(locationJSONString);
    List<TaskLocationParam> taskList = new ArrayList<>();
    for (int i = 0; i < jsonArray.size(); i++) {
      JSONObject jsonObject = jsonArray.getJSONObject(i);
      long taskCode = jsonObject.getLong("taskCode");
      int x = jsonObject.getInt("x");
      int y = jsonObject.getInt("y");
      taskList.add(new TaskLocationParam(taskCode, x, y));
    }
    this.setLocations(taskList);
  }
}
